package com.renan.csv;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.bean.copier.CopyOptions;
import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.date.DateTime;
import cn.hutool.core.date.DateUnit;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.io.FileTypeUtil;
import cn.hutool.core.io.FileUtil;
import cn.hutool.core.io.file.FileNameUtil;
import cn.hutool.core.map.MapUtil;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.core.util.ZipUtil;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
import com.renan.util.CompressionAndDecompressionUtils;
import com.renan.util.DOMUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.dom4j.Document;
import org.dom4j.Element;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.domain.Example;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.io.File;
import java.io.IOException;
import java.nio.file.*;
import java.time.LocalDateTime;
import java.util.*;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
import java.util.stream.Stream;

/**
 * @author 张亚林
 * @date 2020-11-24 文件解压解析入库
 */
@Slf4j
@Component
public class SchedulerTask {

    /**
     * 压缩文件源目录
     */
    @Value("${zip.file.absolute.source.path:/home/zsj/MRO3}")
    private String filePath;

    /**
     * CSV文件加工目录
     */
    @Value("${csv.file.absolute.target.path:/home/zsj/MRO_CSV3}")
    private String csvPath;

    /**
     * 待解压文件扩展名
     */
    @Value("${file.source.type:zip,gz}")
    private String fileType;

    /**
     * 待解析的文件扩展名
     */
    @Value("${file.target.type:xml}")
    private String targetType;

    @Resource(name = "mroRepository")
    private MroRepository mroRepository;


    /**
     * 任务调度执行·自定义线程池
     * 第一次执行 延后一小时；后续每天执行一次
     */
    @Async("taskExecutor")
    @Scheduled(cron = "0 0/5 * * * ?")
    public void task() {
        String msg = "定时器开始执行，执行时间戳【" + DateUtil.now() + "】";
        log.warn("当前线程【{}】,时间【{}】开始执行批量入库!", Thread.currentThread().getName(), DateUtil.now());
        if (StringUtils.isNotBlank(filePath)) {
            // 目录不为空 遍历目录下文件(只帅选zip gz结尾文件)
            try {
                //过滤非xml 或ZIP文件
                DirectoryStream.Filter<Path> notXmlORZipFilter = path -> {
                    final File xfile = path.toFile();
                    if (xfile.exists() && xfile.isFile()) {
                        final String xtype = FileTypeUtil.getType(xfile);
                        return StringUtils.containsAny(xtype, "xml", "gz", "zip");
                    }
                    return false;
                };
                //待读取文件夹(ZIP文件夹)
                final DirectoryStream<Path> dirStream = Files.newDirectoryStream(Paths.get(filePath), notXmlORZipFilter);
                final DirectoryStream<Path> paths = Files.newDirectoryStream(Paths.get(csvPath), notXmlORZipFilter);
                final Iterator<Path> iteratorZIP = dirStream.iterator();
                final Iterator<Path> iteratorCsv = paths.iterator();
                List<Path> zipList = Lists.newArrayList(iteratorZIP);
                List<Path> myList = Lists.newArrayList(iteratorCsv);
                Set<Path> zipSets = Sets.newHashSet(new HashSet(zipList));
                Set<Path> csvSets = Sets.newHashSet(new HashSet(myList));
                //交集  已处理的文件列表
                Set<String> collectZip = zipSets.stream().map(Path::toFile).map(File::getName).map(x -> StrUtil.subBefore(x, ".", true)).collect(Collectors.toSet());
                Set<String> collectCsv = csvSets.stream().map(Path::toFile).map(File::getName).map(x -> StrUtil.subBefore(x, ".", true)).collect(Collectors.toSet());
                Sets.SetView<String> intersection = Sets.intersection(collectZip, collectCsv);
                Sets.SetView<String> diff = Sets.difference(collectZip, collectCsv);
                log.info("当前时间【{}】,线程【{}】,已处理的文件个数为【{}】个!,待处理的文件个数为【{}】个!", LocalDateTime.now(), Thread.currentThread().getName(), intersection.size(), diff.size());
                zipList.stream().parallel().forEach(file -> {
                    final File sourceFile = file.toFile();
                    final File targetFile = FileUtil.file(csvPath, sourceFile.getName());
                    final String type = FileTypeUtil.getType(sourceFile);
                    final long size = sourceFile.length();
                    log.info("线程【{}】,当前遍历文件名【{}】,线程【{}】,文件类型【{}】,大小:【{}KB】", Thread.currentThread().getName(), file.getFileName(), Thread.currentThread().getName(), type, size / 1024);
                    if (size > 0 && StrUtil.containsAnyIgnoreCase(type, StrUtil.splitToArray(fileType, ','))) {
                        // 移动文件 并重命名源文件
                        final String ssuffix = FileUtil.getSuffix(sourceFile);
                        final String tsuffix = FileUtil.getSuffix(targetFile);
                        if (StringUtils.equals(ssuffix, tsuffix) && StrUtil.containsAnyIgnoreCase(ssuffix, StrUtil.splitToArray(fileType, ','))) {
                            FileUtil.copyFile(sourceFile, targetFile, StandardCopyOption.COPY_ATTRIBUTES);
                            FileUtil.rename(sourceFile, FileNameUtil.getPrefix(sourceFile) + ".TMP", true);
                            // 开始解压目标文件 返回解压到的目录
                            File unzipFile = null;
                            if (StrUtil.endWithIgnoreCase("zip", ssuffix)) {
                                unzipFile = ZipUtil.unzip(targetFile.getAbsolutePath(), StrUtil.join("/", csvPath, FileNameUtil.getPrefix(targetFile)));
                            } else if (StrUtil.endWithIgnoreCase("gz", ssuffix)) {
                                //linux 解压gz文件
                                try {
                                    //以下方法解压出错或无法解压
                                    /**  FileInputStream fis = new FileInputStream(targetFile);
                                     GZIPInputStream gis = new GZIPInputStream(new BufferedInputStream(fis));
                                     ZipUtil.unGzip(gis);
                                     ZipUtils.unTarGz(targetFile, StrUtil.join("\\", csvPath, FileNameUtil.getPrefix(targetFile)));
                                     */
                                    final String name = FileUtil.getName(targetFile);
                                    final String folder = StrUtil.subBefore(name, '.', false);
                                    log.info("获取到的文件名称【{}】,目录名称【{}】", name, folder);
                                    //创建文件夹
                                    final File gzdir = FileUtil.mkdir(StrUtil.join("/", csvPath, folder));
                                    final boolean b = CompressionAndDecompressionUtils.gzipDecompression(targetFile.getAbsolutePath(), StrUtil.join("/", csvPath, folder, StrUtil.join(".", folder, targetType)));
                                    if (b) {
                                        unzipFile = gzdir;
                                    }
                                } catch (IOException e) {
                                    e.printStackTrace();
                                }
                            } else {
                                log.error("获取到错误的文件类型【{}】，文件名【{}】", ssuffix, sourceFile.getAbsolutePath());
                            }
                            if (FileUtil.exist(unzipFile)) {
                                // 文件存在 遍历文件
                                final File[] ls = FileUtil.ls(unzipFile.getAbsolutePath());
                                List<Mro> list = Lists.newArrayList();
                                if (ObjectUtil.isNotNull(ls)) {
                                    Stream.of(ls).filter(item -> FileTypeUtil.getType(item).endsWith(targetType)).forEach(x -> {
                                        String xml = "";
                                        try {
                                            log.info("线程【{}】,遍历到的文件路径【{}】,文件名【{}】,文件大小【{}KB】", Thread.currentThread().getName(), x.getAbsolutePath(), x.getAbsoluteFile(), x.length() / 1024);
                                            // 过滤  解析xml文件
                                            Mro mro = new Mro();
                                            log.info("线程【{}】,遍历到的xml文件【{}】", Thread.currentThread().getName(), x.getAbsolutePath());
                                            final Document document = DOMUtils.getXMLByFilePath(x.getAbsolutePath());
                                            xml = document.asXML();
                                            final Element rootElement = document.getRootElement();
                                            final String reversion = rootElement.element("fileHeader").attributeValue("fileFormatVersion");
                                            mro.setFlie(sourceFile.getName());
                                            mro.setFileFormatVersion(reversion);
                                            String endoebId = rootElement.element("eNB").attributeValue("id");
                                            log.info("线程【{}】,获取到基站ID【{}】", Thread.currentThread().getName(), endoebId);
                                            mro.setEnodebId(endoebId);
                                            // 获取smr
                                            final Element element = rootElement.element("eNB").element("measurement").element("smr");
                                            final Iterator<Element> iterator = rootElement.element("eNB").element("measurement").elementIterator("object");
                                            while (iterator.hasNext()) {
                                                final Mro clone = ObjectUtil.clone(mro);
                                                // 遍历Object
                                                final Element next = iterator.next();
                                                final String eci = next.attributeValue("id");
                                                final String mmeUeS1apId = next.attributeValue("MmeUeS1apId");
                                                final String mmeCode = next.attributeValue("MmeCode");
                                                final String timeStamp = next.attributeValue("TimeStamp");
                                                clone.setEci(eci);
                                                clone.setMmeUeS1apId(mmeUeS1apId);
                                                clone.setMmeCode(mmeCode);
                                                clone.setTimeStamp(timeStamp);
                                                // 遍历v节点
                                                final Iterator<Element> iterates = next.elementIterator("v");
                                                final String smrTxt = element.getText();
                                                while (iterates.hasNext()) {
                                                    final Mro cloneV = ObjectUtil.clone(clone);
                                                    final Element nextv = iterates.next();
                                                    final String vtext = nextv.getText();
                                                    if (StringUtils.isNotBlank(vtext)) {
                                                        final String[] split = StringUtils.split(vtext, " ");
                                                        final String[] xmrArr = StringUtils.split(smrTxt, " ");
                                                        if (ObjectUtil.isNotNull(split) && ObjectUtil.isNotNull(xmrArr)) {
                                                            Map<String, String> smrMap = IntStream.range(0, split.length).collect(HashMap::new, (m, i) -> m.put(split[i], xmrArr[i]), (m1, m2) -> {
                                                            });
                                                            if (MapUtil.isNotEmpty(smrMap)) {
                                                                smrMap = MapUtil.reverse(smrMap);
                                                                Map<String, String> filedMapping = Maps.newLinkedHashMap();
                                                                filedMapping.put("MR.LteScEarfcn", "lteScEarfcn");
                                                                filedMapping.put("MR.LteScPci", "lteScPci");
                                                                filedMapping.put("MR.LteScRSRP", "lteScRsrp");
                                                                filedMapping.put("MR.LteScRSRQ", "lteScRsrq");
                                                                filedMapping.put("MR.LteScTadv", "lteScTadv");
                                                                filedMapping.put("MR.LteScPHR", "lteScPhr");
                                                                filedMapping.put("MR.LteScAOA", "lteScAoa");
                                                                filedMapping.put("MR.LteScSinrUL", "lteScSinrUl");
                                                                filedMapping.put("MR.LteScRI1", "lteScRi1");
                                                                filedMapping.put("MR.LteScRI2", "lteScRi2");
                                                                filedMapping.put("MR.LteScRI4", "lteScRi4");
                                                                filedMapping.put("MR.LteScRI8", "lteScRi8");
                                                                filedMapping.put("MR.LteSceNBRxTxTimeDiff", "lteScEnbRxTxTimeDiff");
                                                                filedMapping.put("MR.LteScBSR", "lteScBsr");
                                                                filedMapping.put("MR.LteScPUSCHPRBNum", "lteScPuschPrbNum");
                                                                filedMapping.put("MR.LteScPDSCHPRBNum", "lteScPdschPrbNum");
                                                                filedMapping.put("MR.LteNcEarfcn", "lteNcEarfcn");
                                                                filedMapping.put("MR.LteNcPci", "lteNcPci");
                                                                filedMapping.put("MR.LteNcRSRP", "lteNcRsrp");
                                                                filedMapping.put("MR.LteNcRSRQ", "lteNcRsrq");
                                                                filedMapping.put("MR.TdsNcellUarfcn", "tdsNcellUarfcn");
                                                                filedMapping.put("MR.TdsCellParameterId", "tdsCellParameterId");
                                                                filedMapping.put("MR.TdsPccpchRSCP", "tdsPccpchRsrp");
                                                                filedMapping.put("MR.GsmNcellNcc", "gsmNcellNcc");
                                                                filedMapping.put("MR.GsmNcellBcc", "gsmNcellBcc");
                                                                filedMapping.put("MR.GsmNcellBcch", "gsmNcellBcch");
                                                                filedMapping.put("MR.GsmNcellCarrierRSSI", "gsmNcellCarrierRssi");
                                                                BeanUtil.fillBeanWithMap(smrMap, cloneV, CopyOptions.create().setFieldMapping(filedMapping).setIgnoreNullValue(true).setIgnoreError(true).setIgnoreCase(true).setTransientSupport(true));
                                                            }
                                                        }
                                                    }
                                                    list.add(cloneV);
                                                }
                                            }
                                            if (CollUtil.isNotEmpty(list)) {
                                                final DateTime start = DateTime.now();
                                                log.info("线程【{}】,文件名【{}】遍历到的对象数据量【{}】条,批量入库开始，开始时间【{}】", Thread.currentThread().getName(), x.getName(), list.size(), DateUtil.now());
                                                if (list.size() > 1000) {
                                                    List<List<Mro>> lists = Lists.partition(list, 1000);
                                                    lists.parallelStream().peek(e -> log.info("线程【{}】,时间【{}】,累计数据量【{}】条,当前插入数据量【{}】条", Thread.currentThread().getName(), DateUtil.now(), list.size(), e.size())).forEach(ll -> this.mroRepository.saveAll(ll));
                                                } else {
                                                    final List<Mro> mros = this.mroRepository.saveAll(list);
                                                    log.info("线程【{}】,当前文件名【{}】,插入记录条数【{}】", Thread.currentThread().getName(), x.getAbsolutePath(), mros.size());
                                                }
                                                list.clear();
                                                Mro exp = new Mro();
                                                exp.setFlie(sourceFile.getName());
                                                Example<Mro> example = Example.of(exp);
                                                final long count = this.mroRepository.count(example);
                                                final DateTime end = DateTime.now();
                                                log.info("线程【{}】,文件名【{}】遍历到的对象数据量【{}】条,批量入库开始，结束时间【{}】", Thread.currentThread().getName(), x.getName(), list.size(), DateUtil.now());
                                                log.info("线程【{}】,文件名【{}】批量插入【{}】条,消耗时间【{}ms】,最终插入数据量【{}】条!", Thread.currentThread().getName(), x.getName(), list.size(), DateUtil.between(start, end, DateUnit.MS), count);
                                            } else {
                                                log.warn("线程【{}】,文件名【{}】的内容为空,遍历入库掠过！", Thread.currentThread().getName(), x.getName());
                                            }
                                        } catch (Exception e) {
                                            log.error("线程【{}】,读取到不正确文件【{}】,错误信息【{}】,文件名【{}】", Thread.currentThread().getName(), x.getAbsolutePath(), e.getMessage(), x.getName());
                                            e.printStackTrace();
                                        }
                                    });
                                }
                            }
                        } else {
                            log.error("线程【{}】,文件路径【{}】,文件名【{}】已被处理!!!", Thread.currentThread().getName(), targetFile.getAbsolutePath(), FileUtil.getName(targetFile));
                        }
                    }
                });
                dirStream.close();
            } catch (Exception e) {
                e.printStackTrace();
            }
            log.info("线程【{}】,{},源文件路径:{},csv路径:{}", Thread.currentThread().getName(), msg, this.filePath, this.csvPath);
            log.warn("当前线程【{}】,时间【{}】结束执行批量入库!", Thread.currentThread().getName(), DateUtil.now());
        }
    }
}
